package com.sh.data.engine.domain.shims.hive.manager;

import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.sh.data.engine.domain.shims.db.DbOptions;
import com.sh.data.engine.domain.shims.db.GenericDbManager;
import com.sh.data.engine.domain.shims.db.model.FieldInfoDomain;
import com.sh.data.engine.domain.shims.db.model.PreviewDataDomain;
import com.sh.data.engine.domain.shims.db.model.TableInfoDomain;
import com.sh.data.engine.domain.shims.db.model.TableSizeDomain;
import com.sh.data.engine.domain.shims.hive.enums.HiveSiteKeyEnum;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
import org.apache.hadoop.hive.metastore.api.Table;
import org.apache.thrift.TException;
import org.dom4j.Document;
import org.dom4j.DocumentException;
import org.dom4j.Element;
import org.dom4j.io.SAXReader;

import java.io.File;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author: mengzheng.mhc
 * @date: 2024/6/7 14:18
 */
@Slf4j
public class HiveManager extends GenericDbManager {

    private HiveMetaStoreClient hiveMetaStoreClient;

    private static final long Hive_CONNECT_TIMEOUT = 10L;

    public HiveManager(DbOptions options) {
        super(options);
        //    final File file = new File(options.getHiveSiteXmlPath());
        final File file = new File(options.getHiveSiteXmlPath());
        if (!file.exists()) {
            throw new IllegalArgumentException("hive数据源未配置hive-site.xml或hive-site.xml文件不存在");
        }

        try {
            HiveConf.setHiveSiteLocation(file.toURL());

            HiveConf hiveConf = new HiveConf();

            // hive timeout
            long defaultTimeOutVar =
                hiveConf.getTimeVar(HiveConf.ConfVars.METASTORE_CLIENT_SOCKET_TIMEOUT, TimeUnit.SECONDS);
            log.info("hive site defaultTimeOutVar: " + defaultTimeOutVar);
            long minTimeOutVar = Math.min(Hive_CONNECT_TIMEOUT, defaultTimeOutVar);
            hiveConf.setTimeVar(
                HiveConf.ConfVars.METASTORE_CLIENT_SOCKET_TIMEOUT, minTimeOutVar, TimeUnit.SECONDS);
            // retry limit
            //      hiveConf.setIntVar(HiveConf.ConfVars.METASTORETHRIFTCONNECTIONRETRIES, 2);

            hiveMetaStoreClient = new HiveMetaStoreClient(hiveConf);
        } catch (Exception e) {
            log.error("init HiveMetaStoreClient error: {}", e.getMessage());
            throw new RuntimeException("初始化HiveMetaStoreClient异常: 请检查hive-site文件中的配置");
        }
    }

    static {
        String driverClass = "org.apache.hive.jdbc.HiveDriver";
        try {
            Class.forName(driverClass);
        } catch (ClassNotFoundException e) {
            throw new RuntimeException("Could not load db driver class: " + driverClass);
        }
    }

    @Override
    public void testConnection() {
        try {
            this.hiveMetaStoreClient.getAllDatabases();
        } catch (Exception e) {
            throw new RuntimeException(e.getMessage());
        } finally {
            closeClient();
        }
    }

    @Override
    public String getQueryVersionSql() {
        return "SELECT VERSION()";
    }

    @Override
    public boolean hasSchema() {
        return false;
    }

    @Override
    public List<TableInfoDomain> getTableList(String dbName) throws SQLException {
        return getTableMeta(dbName, Lists.newArrayList("MANAGED_TABLE", "EXTERNAL_TABLE"));
    }

    @Override
    public List<TableInfoDomain> getTableList(String dbName, String schema) throws SQLException {
        return getTableList(dbName);
    }

    @Override
    public List<TableInfoDomain> getViewList(String dbName) throws SQLException {
        return getTableMeta(dbName, Lists.newArrayList("VIRTUAL_VIEW"));
    }

    @Override
    public List<TableInfoDomain> getViewList(String dbName, String schema) throws SQLException {
        return getViewList(dbName);
    }

    @Override
    public List<TableInfoDomain> getTableAndViewList(String dbName) throws SQLException {
        return getTableMeta(
            dbName, Lists.newArrayList("MANAGED_TABLE", "EXTERNAL_TABLE", "VIRTUAL_VIEW"));
    }

    @Override
    public List<TableInfoDomain> getTableAndViewList(String dbName, String schema)
        throws SQLException {
        return getTableAndViewList(dbName);
    }

    public List<TableInfoDomain> getTableMeta(String dbName, List<String> tableTypes)
        throws SQLException {
        try {
            final List<String> allTables = hiveMetaStoreClient.getAllTables(dbName);

            if (CollectionUtils.isEmpty(allTables)) {
                return Lists.newArrayList();
            }

            final List<Table> tables = hiveMetaStoreClient.getTableObjectsByName(dbName, allTables);

            List<TableInfoDomain> domainList = new ArrayList<>();

            tables.forEach(
                t -> {
                    final String comment = t.getParameters().get("comment");
                    final String transientLastDdlTime = t.getParameters().get("transient_lastDdlTime");
                    String tableType = t.getTableType();

                    if (!tableTypes.contains(tableType)) {
                        return;
                    }

                    Long lastDdlTime =
                        StringUtils.isBlank(transientLastDdlTime)
                            ? null
                            : Long.valueOf(transientLastDdlTime + "000");

                    if (Objects.equals("VIRTUAL_VIEW", tableType)) {
                        tableType = "VIEW";
                    }

                    final TableInfoDomain tableInfoDomain =
                        TableInfoDomain.builder()
                            .tableName(t.getTableName())
                            .dbName(t.getDbName())
                            .tableComment(comment)
                            .tblType(TableInfoDomain.TblType.from(tableType))
                            .isPartitionTable(t.getPartitionKeysSize() > 0)
                            .lastDdlTime(lastDdlTime)
                            .build();
                    domainList.add(tableInfoDomain);
                });

            return domainList;

        } catch (Exception e) {
            log.error(e.getMessage(), e);
            throw new SQLException("获取表列表异常：" + e.getMessage());
        }
    }

    @Override
    public TableInfoDomain getTableInfoByTableName(String dbName, String tableName)
        throws SQLException {

        tableName = StringUtils.replace(tableName, "`", "");

        try {
            final Table table = hiveMetaStoreClient.getTable(dbName, tableName);
            if (null == table) {
                return null;
            }

            final String comment = table.getParameters().get("comment");
            final String transientLastDdlTime = table.getParameters().get("transient_lastDdlTime");

            Long lastDdlTime =
                StringUtils.isBlank(transientLastDdlTime)
                    ? null
                    : Long.valueOf(transientLastDdlTime + "000");

            final TableInfoDomain tableInfoDomain =
                TableInfoDomain.builder()
                    .dbName(dbName)
                    .schema(dbName)
                    .tableName(tableName)
                    .tableComment(comment)
                    .tblType(TableInfoDomain.TblType.from(table.getTableType()))
                    .lastDdlTime(lastDdlTime)
                    .isPartitionTable(table.getPartitionKeysSize() > 0)
                    .build();

            return tableInfoDomain;
        } catch (NoSuchObjectException e) {
            // 表不存在的时候不抛异常, 和sql异常区分开
            log.info("表不存在：" + e.getMessage());
            return null;
        } catch (TException e) {
            log.error(e.getMessage(), e);
            throw new SQLException("获取表信息异常：" + e.getMessage());
        }
    }

    @Override
    public TableInfoDomain getTableInfoByTableName(String dbName, String schema, String tableName)
        throws SQLException {
        return getTableInfoByTableName(dbName, tableName);
    }

    @Override
    public List<FieldInfoDomain> getFieldList(String dbName, String tableName) throws SQLException {
        List<FieldInfoDomain> fieldInfoDomains = new ArrayList<>();

        tableName = StringUtils.replace(tableName, "`", "");

        try {
            final Table table = hiveMetaStoreClient.getTable(dbName, tableName);

            if (null == table) {
                return fieldInfoDomains;
            }
            final StorageDescriptor sd = table.getSd();
            final List<FieldSchema> partitionKeys = table.getPartitionKeys();

            for (FieldSchema col : sd.getCols()) {
                FieldInfoDomain fieldInfo =
                    FieldInfoDomain.builder()
                        .fieldName(col.getName())
                        .fieldType(col.getType())
                        .comment(col.getComment())
                        .isPartition(false)
                        .isNullable(true)
                        .build();
                fieldInfoDomains.add(fieldInfo);
            }

            for (FieldSchema partitionKey : partitionKeys) {
                FieldInfoDomain fieldInfo =
                    FieldInfoDomain.builder()
                        .fieldName(partitionKey.getName())
                        .fieldType(partitionKey.getType())
                        .comment(partitionKey.getComment())
                        .isPartition(true)
                        .build();
                fieldInfoDomains.add(fieldInfo);
            }

            return fieldInfoDomains;
        } catch (TException e) {
            log.error(e.getMessage(), e);
            throw new SQLException("获取字段信息异常：" + e.getMessage());
        }
    }

    @Override
    public List<FieldInfoDomain> getFieldList(String dbName, String schema, String tableName)
        throws SQLException {
        return getFieldList(dbName, tableName);
    }

    @Override
    public List<TableSizeDomain> getTableSizeList(String dbName, String schema) throws SQLException {
        try {
            final List<String> allTables = hiveMetaStoreClient.getAllTables(dbName);
            if (CollectionUtils.isEmpty(allTables)) {
                return Collections.EMPTY_LIST;
            }

            final List<Table> tables = hiveMetaStoreClient.getTableObjectsByName(dbName, allTables);

            if (CollectionUtils.isEmpty(tables)) {
                return Collections.EMPTY_LIST;
            }

            return tables.stream()
                .map(
                    t -> {
                        final Long totalSize =
                            Long.valueOf(t.getParameters().getOrDefault("totalSize", "0"));
                        TableSizeDomain tableSizeDomain = new TableSizeDomain();
                        tableSizeDomain.setTableSize(totalSize);
                        tableSizeDomain.setTableName(t.getTableName());
                        return tableSizeDomain;
                    })
                .collect(Collectors.toList());
        } catch (TException e) {
            throw new SQLException("获取表列表异常：" + e.getMessage());
        } finally {
            closeClient();
        }
    }

    @Override
    public boolean alterFieldDesc(
        String dbName, String tableName, String fieldName, String fieldDes, boolean isPartition)
        throws SQLException {
        //    String hiveSiteXmlPath = this.dbOptions.getHiveSiteXmlPath();
        //    Map<String, String> mateXmlInfo = getMateXmlInfo(hiveSiteXmlPath);
        //    // 1.创建基础数据源
        //    BasicDataSource bds = new BasicDataSource();
        //    // 2.设置属性
        //    bds.setUrl(mateXmlInfo.get(HiveSiteKeyEnum.METASORE_JDBC_URL.getKey()));
        //    bds.setUsername(mateXmlInfo.get(HiveSiteKeyEnum.METASORE_JDBC_USERNAME.getKey()));
        //    bds.setPassword(mateXmlInfo.get(HiveSiteKeyEnum.METASORE_JDBC_PASSWORD.getKey()));
        //    Connection connection = bds.getConnection();
        //
        //      String sql;
        //      if (isPartition) {
        //        sql =
        //                "UPDATE PARTITION_KEYS "
        //                        + " JOIN TBLS ON PARTITION_KEYS.TBL_ID = TBLS.TBL_ID "
        //                        + " JOIN DBS ON TBLS.DB_ID = DBS.DB_ID "
        //                        + " SET PARTITION_KEYS.PKEY_COMMENT = '"
        //                        + fieldDes
        //                        + "' WHERE "
        //                        + " DBS. NAME = '"
        //                        + dbName
        //                        + "' AND TBLS.TBL_NAME = '"
        //                        + tableName
        //                        + "' AND PARTITION_KEYS.PKEY_NAME = '"
        //                        + fieldName
        //                        + "'";
        //      } else {
        //        sql =
        //                "UPDATE COLUMNS_V2 JOIN SDS ON COLUMNS_V2.CD_ID = SDS.CD_ID "
        //                        + " JOIN TBLS ON SDS.SD_ID = TBLS.SD_ID "
        //                        + " JOIN DBS ON TBLS.DB_ID = DBS.DB_ID "
        //                        + " SET `COMMENT`= '"
        //                        + fieldDes
        //                        + "' "
        //                        + " WHERE DBS.NAME = '"
        //                        + dbName
        //                        + "' AND  TBLS.TBL_NAME = '"
        //                        + tableName
        //                        + "' AND COLUMNS_V2.COLUMN_NAME= '"
        //                        + fieldName
        //                        + "' ";
        //      }
        //    try{
        //      connection.createStatement().execute(sql);
        //      return true;
        //    }catch (Exception e){
        //
        //    }finally{
        //      connection.close();
        //    }

        throw new SQLException("暂未实现");
    }

    @Override
    public PreviewDataDomain previewData(String tableName) throws SQLException {
        final PreviewDataDomain previewDataDomain = super.previewData(tableName);
        final List<String> fieldNameList = previewDataDomain.getFieldNameList();
        for (int i = 0; i < fieldNameList.size(); i++) {
            String fieldName = StringUtils.substringAfter(fieldNameList.get(i), ".");
            fieldNameList.set(i, fieldName);
        }
        return previewDataDomain;
    }

    /**
     * 获取hive元数据连接信息url、username、password
     *
     * @param hiveSiteXmlPath
     * @return
     */
    public Map<String, String> getMetaXmlInfo(String hiveSiteXmlPath) {
        // 1.创建Reader对象
        SAXReader reader = new SAXReader();
        // 2.加载xml
        Document document = null;
        try {
            document = reader.read(new File(hiveSiteXmlPath));
        } catch (DocumentException e) {

        }
        HashMap<String, String> map = Maps.newHashMap();
        // 3.获取根节点
        Element rootElement = document.getRootElement();
        Iterator iterator = rootElement.elementIterator();
        while (iterator.hasNext()) {
            Element stu = (Element) iterator.next();
            Iterator iterator1 = stu.elementIterator();

            while (iterator1.hasNext()) {
                Element stuChild = (Element) iterator1.next();
                if (HiveSiteKeyEnum.METASTORE_JDBC_URL.getKey().equals(stuChild.getStringValue())) {
                    map.put(
                        HiveSiteKeyEnum.METASTORE_JDBC_URL.getKey(),
                        ((Element) iterator1.next()).getStringValue());
                }
                if (HiveSiteKeyEnum.METASTORE_JDBC_USERNAME.getKey().equals(stuChild.getStringValue())) {
                    map.put(
                        HiveSiteKeyEnum.METASTORE_JDBC_USERNAME.getKey(),
                        ((Element) iterator1.next()).getStringValue());
                }
                if (HiveSiteKeyEnum.METASTORE_JDBC_PASSWORD.getKey().equals(stuChild.getStringValue())) {
                    map.put(
                        HiveSiteKeyEnum.METASTORE_JDBC_PASSWORD.getKey(),
                        ((Element) iterator1.next()).getStringValue());
                }
                if (HiveSiteKeyEnum.METASTORE_URIS.getKey().equals(stuChild.getStringValue())) {
                    map.put(
                        HiveSiteKeyEnum.METASTORE_URIS.getKey(),
                        ((Element) iterator1.next()).getStringValue());
                }
            }
        }
        return map;
    }

    /**
     * HiveManager内部关闭client
     */
    private void closeClient() {
        if (null != this.hiveMetaStoreClient) {
            log.info("关闭hiveMetaStoreClient");
            this.hiveMetaStoreClient.close();
        }
    }

    /**
     * 关闭client
     *
     * @param closeables
     */
    @Override
    public void close(AutoCloseable... closeables) {
        super.close(closeables);
        if (null != this.hiveMetaStoreClient) {
            log.info("关闭hiveMetaStoreClient");
            this.hiveMetaStoreClient.close();
        }
    }

    public static void main(String[] args) throws Exception {
        // Map mateXmlInfo = getMateXmlInfo("");
        // System.out.println(mateXmlInfo);

        DbOptions dbOptions = new DbOptions();
        dbOptions.setUrl("jdbc:hive2://10.88.36.217:10000/default");
        dbOptions.setUsername("hive");
        dbOptions.setPassword("");
        dbOptions.setHiveSiteXmlPath("/Users/fuchen/Downloads/hive-site.xml");
        HiveManager hiveManager = new HiveManager(dbOptions);

        String sql = "add jar /home/hiveudf-1.0-SNAPSHOT.jar";
        hiveManager.queryForString(sql);

        String sql2 = "create temporary function test as 'test.Func'";
        hiveManager.queryForString(sql2);

        String rsSql = "select test(id)from ods_table0606";
        List<String> strings = hiveManager.queryForString(rsSql);

        //    final TableInfoDomain aDefault = hiveManager.getTableInfoByTableName("default", table);
        System.out.println(strings);
    }
}